package com.sarnath.sardoop.modules.hbase.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

import javax.annotation.Resource;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.ResultsExtractor;
import org.springframework.stereotype.Service;

import com.sarnath.sardoop.common.utils.StringUtils;
import com.sarnath.sardoop.modules.hbase.common.HQueryFilterUtil;
import com.sarnath.sardoop.modules.hbase.entity.QueryCondition;
import com.sarnath.sardoop.modules.hbase.reflect.HbaseBeanConverter;

@Service("hbaseQuery")
@Lazy(false)
public class HbaseQueryService {
	@Resource
	private HbaseTemplate htemplate;
	
	/**
	 * condition中的RowEntityClass包含实体数据的映射关系
	 * @param condition 查询条件
	 * @return 映射后的实体列表
	 */
	public List<Object> query(final QueryCondition condition){
		final HbaseBeanConverter convert=new HbaseBeanConverter(condition.getRowEntityClass());
		return htemplate.find(condition.getTableName(), initScan(condition), new ResultsExtractor<List<Object>>() {
			public List<Object> extractData(ResultScanner results) throws Exception{
				List<Object> objList = new ArrayList<>();
				Result rs =results.next();
				while(rs !=null){
					Object bean=condition.getRowEntityClass().newInstance();
					convert.convert(bean, rs.getRow());
					NavigableMap<byte[], NavigableMap<byte[], byte[]>>  rowMap=rs.getNoVersionMap();
					for(byte[] f:rowMap.keySet()){
						NavigableMap<byte[], byte[]> colMap=rowMap.get(f);
						for(byte[] c:colMap.keySet()){
							convert.convert(bean,Bytes.toString(f),Bytes.toString(c),colMap.get(c));
						}
					}
					objList.add(bean);
					rs =results.next();
				}
				return objList;
			}
		});
	}
	
	/**
	 * condition中的RowEntityClass包含实体数据的映射关系
	 * @param condition 查询条件
	 * @return <rowkey,columnMap>
	 */
	public Map<String,Map<String,String>> queryMap(final QueryCondition condition){
		return htemplate.find(condition.getTableName(), initScan(condition), new ResultsExtractor<Map<String,Map<String,String>>>() {
			public Map<String,Map<String,String>> extractData(ResultScanner results) throws Exception{
				Map<String,Map<String,String>> objMap = new HashMap<String, Map<String,String>>();
				Result rs =results.next();
				while(rs !=null){
					Map<String,String> rowMap=new HashMap<>();
					objMap.put(Bytes.toString(rs.getRow()), rowMap);
					NavigableMap<byte[], NavigableMap<byte[], byte[]>>  rowMap1=rs.getNoVersionMap();
					for(byte[] f:rowMap1.keySet()){
						NavigableMap<byte[], byte[]> colMap=rowMap1.get(f);
						for(byte[] c:colMap.keySet()){
							rowMap.put(Bytes.toString(c), Bytes.toString(colMap.get(c)));
						}
					}
					rs =results.next();
				}
				return objMap;
			}
		});
	}
	
	/**
	 * @param condition 查询条件
	 * @return 设置好查询条件的Scan实例
	 */
	private Scan initScan(QueryCondition condition){
		Scan scan = new Scan();
		scan.setCaching(condition.getCacheSize());
		scan.setCacheBlocks(condition.isCacheBlocks());
		if(StringUtils.isNotBlank(condition.getRowPrefix())){
			scan.setRowPrefixFilter(condition.getRowPrefix().getBytes());
		}
		List<Filter> filterList=new ArrayList<>();
		if(condition.getFamily()!=null){
			for(int i=0;i<condition.getFamily().length;i++){
				filterList.add(HQueryFilterUtil.newSubStringFilter(condition.getFamily()[i], condition.getField()[i], condition.getValue()[i]));
			}
		}
		if(condition.getRowLimit()>0){//行数限制在最后
			filterList.add(HQueryFilterUtil.rowLimitFilter(condition.getRowLimit()));
		}
		if(filterList.size()>0){
			scan.setFilter(HQueryFilterUtil.and(filterList));
		}
		return scan;
	}
}
